package model
// 引入相关包和训练数据集
import org.apache.spark.sql.SparkSession
import org.apache.spark.ml.feature._
import org.apache.spark.ml.classification.LogisticRegression
import org.apache.spark.ml.{Pipeline,PipelineModel}
import org.apache.spark.ml.linalg.Vector
import org.apache.spark.sql.Row

object LogisticRegression {
  def LogisticReg(): Unit = {
    // 构建一个机器学习pipline工作流
    // stage[0]: Tokenizer.transform（）方法将原始文本文档拆分为单词，向DataFrame添加一个带有单词的新列。
    // stage[1]: HashingTF.transform（）方法将字列转换为特征向量，向这些向量添加一个新列到DataFrame。
    // stage[2]: LogisticRegression.Estimator（）基于概率的词分类

    val spark = SparkSession
      .builder()
      .master("local")
      .appName("pipelines")
      .getOrCreate()
    import spark.implicits._ //和SQLContext一样开启隐式转换

    val training = spark.createDataFrame(Seq(
      (0L, "a b c d e spark", 1.0),
      (1L, "b d", 0.0),
      (2L, "spark f g h", 1.0),
      (3L, "hadoop mapreduce", 0.0)
    )).toDF("id", "text", "label")

    // pipelines包含tokenizer，hashingTF，lr三个步骤
    // 将原始文本拆分成单词
    val tokenizer = new Tokenizer()
      .setInputCol("text")
      .setOutputCol("words")

    //################################################
    println("##############tokenizer#############")
    val testtokenizer = tokenizer.transform(training)

    testtokenizer.show(false)
    //################################################

    // 将单词转化成特征向量
    val hashingTF = new HashingTF()
      .setNumFeatures(2000)
      .setInputCol(tokenizer.getOutputCol)
      .setOutputCol("features")

    //################################################
    println("##############hashingTF#############")
    val testhashingTF = hashingTF.transform(testtokenizer)
    testhashingTF.show(false)
    //################################################

    // 创建一个机器学习模型并设置参数
    val lr = new LogisticRegression()
      .setMaxIter(10)
      .setRegParam(0.01)

    val pipeline = new Pipeline()
    // 建立一个pipeline
      .setStages(Array(tokenizer, hashingTF, lr))

    //通过piprline的fit方法传入训练数据,创建model
    val model = pipeline.fit(training)

    // 测试数据
    val test = spark.createDataFrame(Seq(
      (4L, "spark i j k"),
      (5L, "l m n"),
      (6L, "spark a"),
      (7L, "apache hadoop")
    )).toDF("id", "text")

    // 通过transform方法预测
    val pre = model.transform(test)
    pre.show(false)

    // 看到预测的概率值
    // 由于训练数据集较少，如果有更多的测试数据进行学习，预测的准确率将会有显著提升
    // rawPrediction 是逻辑回归分类器的原始输出（长度等于类数的数组）
    //  probability是将逻辑函数应用于rawPrediction（长度等于 的数组rawPrediction）的结果
    //  prediction是数组probability取最大值的参数，它给出最可能的标签（单个数字）
  }

  def main(args: Array[String]): Unit = {
    LogisticReg()
  }
}
